package com.mini.gateway.core.filter;

import com.alibaba.fastjson.JSON;
import com.mini.gateway.core.modules.common.KafkaService;
import com.mini.gateway.core.modules.log.entity.ApiLog;
import com.mini.gateway.core.modules.log.service.ApiLogConstants;
import com.mini.gateway.utils.DateUtil;
import com.netflix.hystrix.exception.HystrixRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.cloud.gateway.route.Route;
import org.springframework.cloud.gateway.support.TimeoutException;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Mono;

import java.net.InetSocketAddress;
import java.net.URI;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.function.Function;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR;

/**
 * 接口调用信息记录
 *
 */
@Slf4j
@Component
public class ElapsedTimeGatewayFilterFactory extends AbstractGatewayFilterFactory implements Ordered {

    @Autowired
    private KafkaService kafkaService;

    @Override
    public GatewayFilter apply(Object config) {
        return ((exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
            long startTime = System.currentTimeMillis();
            String serialNo = saveLog(route, request, startTime);
            exchange.getAttributes().put(ApiLogConstants.API_LOG_SERIAL, serialNo);
            return chain.filter(exchange).then(
                    Mono.fromRunnable(() -> {
                        try {
                            ApiLog apiLog = new ApiLog();
                            apiLog.setSerialNo(serialNo);
                            apiLog.setResponseCode(exchange.getResponse().getStatusCode().value());
                            apiLog.setEndTime(LocalDateTime.now());
                            String json = JSON.toJSONString(apiLog);
                            // 日志投递到kafka，做记录、预警处理
                            kafkaService.sendMessage(ApiLogConstants.KAFKA_TOPIC_APILOG, serialNo, json);
                        } catch (Exception ex) {
                            log.error("日志存储异常：", ex);
                        }
                    })
            );
        });
    }

    /**
     * 记录调用日志
     * @param route
     * @param request
     * @param startTime
     * @return
     */
    private String saveLog(Route route, ServerHttpRequest request, long startTime) {
        // 唯一序列
        String serialNo = UUID.randomUUID().toString();
        try {
            String method = request.getMethodValue();
            URI uri = request.getURI();
            String host = uri.getHost();
            int port = uri.getPort();
            String path = uri.getPath();
            String remoteHost = null;
            InetSocketAddress remoteAddr = request.getRemoteAddress();
            if (remoteAddr != null) {
                remoteHost = remoteAddr.getHostName();
            }
            ApiLog apiLog = new ApiLog();
            apiLog.setSerialNo(serialNo);
            String routeId = route != null ? route.getId() : null;
            apiLog.setRouteId(routeId);
            apiLog.setMethod(method);
            apiLog.setHost(host);
            apiLog.setPort(port);
            apiLog.setPath(path);
            apiLog.setUri(uri.toString());
            apiLog.setRemoteHost(remoteHost);
            Instant instant = Instant.ofEpochMilli(startTime);
            apiLog.setStartTime(LocalDateTime.ofInstant(instant, DateUtil.ZONE));

            String json = JSON.toJSONString(apiLog);
            // 日志投递到kafka，做记录、预警处理
            kafkaService.sendMessage(ApiLogConstants.KAFKA_TOPIC_APILOG, serialNo, json);
        } catch (Exception ex) {
            log.error("日志存储异常：", ex);
        }
        return serialNo;
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }
}
